package com.leyantech.ecp.agenx.connector.handler;

import static com.leyantech.ecp.agenx.common.redis.RedisKeyUtils.keyOfAssistantConnectLock;

import com.leyantech.ecp.agenx.common.config.Constants;
import com.leyantech.ecp.agenx.common.config.MessageAddress;
import com.leyantech.ecp.agenx.common.misc.TracingHelper;
import com.leyantech.ecp.agenx.common.misc.ZkLockUtils;
import com.leyantech.ecp.agenx.connector.di.WebSocketMessageHandlerFactory;

import com.ctrip.framework.apollo.ConfigService;
import com.google.inject.Inject;
import io.vertx.core.Handler;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.impl.VertxInternal;
import io.vertx.core.impl.future.PromiseInternal;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.auth.User;
import io.vertx.ext.web.RoutingContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-10-27.
 */
@Slf4j
public class WebSocketConnectHandler implements Handler<RoutingContext> {

  private final VertxInternal vertx;

  private final WebSocketMessageHandlerFactory factory;

  @Inject
  public WebSocketConnectHandler(VertxInternal vertx,
      WebSocketMessageHandlerFactory factory) {
    this.vertx = vertx;
    this.factory = factory;
  }

  @Override
  public void handle(RoutingContext context) {

    // 异步处理，当前 http stream 暂停读，等异步完成后再继续读 request.resume()
    final HttpServerRequest request = context.request().pause();

    final User user = context.user();

    final String storeId = user.get(Constants.STORE_ID_FIELD);
    final String assistantId = user.get(Constants.ASSISTANT_ID_FIELD);

    // 确保 (sid+aid) 在集群中只能建立一个长连接
    // 本来打算用 redis 做分布式锁，但考虑到 connector 尽量精简，所以没有依赖 redis, 而是依赖自身的集群管理器 zk
    final String lockName = keyOfAssistantConnectLock(storeId, assistantId);

    ZkLockUtils.getLock(lockName, 0L).onComplete(TracingHelper.tracingHandler(lock -> {
      if (lock.failed()) {
        log.warn("assistantId {} in connecting, ignored", assistantId, lock.cause());
        context.end("in connecting, hold on.");
        return;
      }

      final PromiseInternal<Void> registrationDonePromise = vertx.promise();

      try {
        request.resume()
            .toWebSocket()
            .onComplete(TracingHelper.tracingHandler(result -> {
              if (result.succeeded()) {
                final ServerWebSocket websocket = result.result();
                factory.create(user, registrationDonePromise).handle(websocket);

                vertx.executeBlocking(ignored -> {
                  // 通知该客服的其它链接关闭
                  final String handlerID = websocket.textHandlerID();
                  final String reason = ConfigService.getAppConfig()
                      .getProperty("Substitution.Reason", "KickOut");
                  final JsonObject cmd = new JsonObject()
                      .put("header", new JsonObject().put("fn", "KickOut"))
                      .put("body", new JsonObject()
                          .put("handlerID", handlerID)
                          .put("reason", reason)
                      );

                  final DeliveryOptions deliveryOptions = new DeliveryOptions()
                      .addHeader("type", "cmd");

                  final String address = MessageAddress.toConnector(storeId, assistantId);
                  vertx.eventBus().publish(address, cmd.toString(), deliveryOptions);
                });
              } else {
                final Throwable cause = result.cause();
                registrationDonePromise.fail(cause);
                log.error("upgrade websocket failed", cause);
              }
            }));
      } catch (Exception e) {
        log.warn("upgrade websocket failed", e);
        registrationDonePromise.fail(e);
      }

      registrationDonePromise.onComplete(result -> {
        if (result.failed()) {
          final Throwable cause = result.cause();
          log.warn("connect failed by {} ", ExceptionUtils.getMessage(cause), cause);
          context.fail(cause);
        }

        log.info("connection {}, release lock {}", result.succeeded(), lockName);
        lock.result().release();
      });
    }));

  }

}
